package engine

import (
	"fmt"
	"miniDW/src/log"
	"sync"
)

type Task struct {
	components map[string]*Component
}

func (t *Task) createComponents(cfgs []ConfigInfo) error {
	t.components = make(map[string]*Component)
	var preComponent *Component = nil
	for _, config := range cfgs {
		log.Debug(fmt.Sprintf("Create component: %s", config.Name))
		component, err := CreateComponent(config)
		if err != nil {
			t.Close()
			return err
		}

		if preComponent != nil {
			//处理上游组件
			sources := component.SourceName()
			if len(*sources) == 0 {
				//如果没有配置上游组件，则默认上一个组件为上游组件
				names := make([]string, 0)
				names = append(names, preComponent.Name())
				component.SetSourceName(names)
			}
			//处理上一组件的下游组件
			targets := preComponent.TargetName()
			if len(*targets) == 0 {
				//如果上一组件没有配置下游组件，则默认当前组件为上游组件的下游组件
				names := make([]string, 0)
				names = append(names, component.Name())
				preComponent.SetTargetName(names)
			}
		}

		preComponent = component
		t.components[config.Name] = component
	}
	return nil
}

func (t *Task) connect() error {
	for _, component := range t.components {
		for _, targetName := range *component.TargetName() {
			target, ok := t.components[targetName]
			if !ok {
				return log.Error(fmt.Errorf("target(%s) not exist.", targetName))
			}
			connector := target.GetSourceConnector()
			err := component.AddTargetConnector(connector)
			if err != nil {
				return err
			}
		}

		//在Open中需要创建数据的结构，并传递给下游
		err := component.function.Open(component)
		if err != nil {
			return err
		}
	}
	return nil
}

func CreateTask(cfgs []ConfigInfo) (*Task, error) {
	task := &Task{}

	//创建组件
	err := task.createComponents(cfgs)
	if err != nil {
		return nil, err
	}

	//连接组件
	err = task.connect()
	if err != nil {
		return nil, err
	}

	return task, nil
}

func (t *Task) Close() {
	for _, component := range t.components {
		_ = component.Close()
	}
}

func (t *Task) Run() {
	var wg sync.WaitGroup
	for _, component := range t.components {
		wg.Add(1)
		go component.Run(&wg)
	}
	log.Debug("Waiting for all components to finish.")
	wg.Wait()
	log.Debug("All components finished.")

}
